package com.li.controller;

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @program: srpingboot-rocketmq
 * @description:
 * @author: li
 * @create: 2022-09-23 11:14
 **/
@RestController
@RequestMapping("/rocketmq")
public class TestController {

    private static final String group = "order_product_group_name";
    private static final String addr = "192.168.179.129:9876";

    private static final ThreadPoolExecutor POOL_EXECUTOR = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    private static final Logger logger = LoggerFactory.getLogger(TestController.class);

    @Resource
    private RocketMQTemplate rocketMQTemplate;

//    @Resource
//    private MessageQueueSelector selector;

    @Autowired
    private DefaultMQProducer defaultMQProducer;


    /**
     * 发送普通同步 消息
     *
     * @return
     */
    @RequestMapping("/commonSend")
    public String commonSend() {
        try {
//            //使用默认producer
//            DefaultMQProducer mqProducer = new DefaultMQProducer(group);
//            mqProducer.setNamesrvAddr(addr);
//            mqProducer.start();
            System.out.println("---start---");
            Message message = new Message();
            message.setTopic("TOPIC_TEST");
            message.setBody("hello !!!".getBytes());
            // 用户属性扩展点
            message.putUserProperty("traceid", "666666");
            SendResult sendResult = defaultMQProducer.send(message);
            defaultMQProducer.setDefaultTopicQueueNums(16);//设置topic的 队列数量
            defaultMQProducer.setRetryTimesWhenSendFailed(3);//设置发送失败后重试次数
            defaultMQProducer.setSendLatencyFaultEnable(true); // 开启延迟隔离策略
            System.out.println(sendResult);
            System.out.println("---end---");
            return sendResult.toString();

        } catch (Exception e) {
            logger.error("commonSend", e);
            return "error";
        }
    }

    /**
     * 发送普通同步 消息
     *
     * @return
     */
    @RequestMapping("/commonSendTest")
    public String commonSendTest() throws Exception{
        try {
//            int i = 1/0;
//            //使用默认producer
//            DefaultMQProducer mqProducer = new DefaultMQProducer(group);
//            mqProducer.setNamesrvAddr(addr);
//            mqProducer.start();
            System.out.println("---start---");
            Message message = new Message();
            message.setTopic("TOPIC_TEST_A");
            message.setBody("hello !!!".getBytes());
            SendResult sendResult = defaultMQProducer.send(message);
            defaultMQProducer.setDefaultTopicQueueNums(16);//设置topic的 队列数量
            defaultMQProducer.setRetryTimesWhenSendFailed(3);//设置发送失败后重试次数
            defaultMQProducer.setSendLatencyFaultEnable(true); // 开启延迟隔离策略
            System.out.println(sendResult);
            System.out.println("---end---");
            return sendResult.toString();

        } catch (Exception e) {
            logger.error("commonSend", e);
            throw e;
        }
    }

    /**
     * 发送普通同步 tag消息
     *
     * @return
     */
    @RequestMapping("/commonSendTag")
    public String commonSendTag() {
        try {
//            //使用默认producer
//            DefaultMQProducer mqProducer = new DefaultMQProducer(group);
//            mqProducer.setNamesrvAddr(addr);
//            mqProducer.start();
            System.out.println("---start tag2---");
            Message message = new Message();
            message.setTopic("TOPIC_TEST");
            message.setTags("tag2");
            message.setBody("hello !!!".getBytes());
            SendResult sendResult = defaultMQProducer.send(message);
            defaultMQProducer.setDefaultTopicQueueNums(16);//设置topic的 队列数量
            defaultMQProducer.setRetryTimesWhenSendFailed(3);//设置发送失败后重试次数
//            defaultMQProducer.setRetryTimesWhenSendAsyncFailed();
//            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK();
            defaultMQProducer.setSendLatencyFaultEnable(true); // 开启延迟隔离策略
            System.out.println(sendResult);
            System.out.println("---end---");
            return sendResult.toString();

        } catch (Exception e) {
            logger.error("commonSend", e);
            return "error";
        }
    }
    /**
     * 发送普通异步消息
     *
     * @return
     */
    @RequestMapping("/commonSendAsync")
    public String commonSendAsync() {
        try {

            for (int i = 0; i < 10; i++) {
                Message message = new Message();
                message.setTopic("TOPIC_TEST");
                message.setBody(("hello async " + i).getBytes());
                int index = i + 1;
                POOL_EXECUTOR.submit(() -> {
                    asyncSend(message, index);
                });
            }
            return "success";
        } catch (Exception e) {
            logger.error("commonSend", e);
            return "error";
        }
    }

    public void asyncSend(Message message, int index) {
        try {
            defaultMQProducer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("index :" + index);
                    System.out.println("消息id：" + sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("index :" + index);
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 顺序消息发送， 需要同步发送，保证当前消息发送成功才能发送下一个消息
     * 通过MessageQueueSelector 根据 当前业务key 发送到 同一个消息队列中
     */
    @RequestMapping("/orderSend")
    public void orderSend() {
        try {
            String orderKey = "123456";
            for (int i = 0; i < 10; i++) {
                System.out.println("发送顺序消息");
                Message message = new Message();
                message.setTopic("TOPIC_TEST_ORDER");
                message.setTags("tag_order");
                message.setBody(("orderKey 123456 顺序消息：" + i).getBytes());
                defaultMQProducer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // 传入的orderKey
                        int orderKey = Integer.parseInt(arg.toString());
                        System.out.println("MessageQueueSelector select arg:" + arg);
                        int index = orderKey % mqs.size();
                        System.out.println("MessageQueueSelector select index:" + index);
                        return mqs.get(index);
                    }
                }, orderKey);
            }
        } catch (Exception e) {

        }
    }


    /**
     * 事务消息：
     * 事务消息发送分为两个阶段
     * 第一个阶段为发送一个半事务消息，半事务消息是指暂不能投递的消息，生产者已经成功的将消息发送到broker，但是broker为收到生产者对该消息的二次确认，
     * 此时将当前消息置为"暂不能投递状态"，
     * 第二阶段，半事务消息二次确认，如果发送成功则执行本地事务，并根据本地事务执行成功与否，向broker发送半事务消息的状态(commit或者rollback)，半事务消息只有commit状态才会真正的向下游投递
     * 如果由于网络问题、生产者应用重启等导致某条事务线下二次确认丢失，broker端会通过扫描发现某条消息长期处于 半事务消息，主动向消息生产者询问
     * 该消息的最终状态(commit或者rollback)， 最终保持了本地事务执行成功，下游就能收到消息，本地事务执行失败，下游就不能收到消息
     * <p>
     * <p>
     * 事务消息发送步骤：
     * 1.生产者将半事务消息发送这 rocketmq broker
     * 2.rocketmq broker 将消息持久化成功之后，向生产者返回ACK确认消息已经成功发送，此时当前消息暂不能投递，为半事务消息
     * 3.生产者开始执行本地事务逻辑
     * 4.生产者根据本地事务执行结果向服务器broker 提交二次确认结果(commit或者rollback)，服务端收到确认结果后处理：
     * 二次确认结果为commit，服务端将半事务消息标记为可投递，并投递给消费者
     * 二次确认结果为rollback，服务端将回滚事务，不会将半事务消息投递给消费者
     * 5.在断网或者生产者重启情况下，若服务端未收到消息发送者提交的二次确认结果，或服务端收到确认结果为Unknown未知状态，经过固定时间，服务端将对消息生产者发起消息会查，确认二次提交结果
     * 生产者受到消息回查后，需要检查对应消息的本地事务执行最终结果，然后向broker服务端再次提交二次确认
     * 6.需要注意的是， 服务端仅仅会按照参数尝试指定次数，超过次数后事务会强制回滚，因此未决事务的回查时效性非常关键，需要按照业务的实际风险来设置
     */
    @RequestMapping("/transSend")
    public void transSend() {
        try {
            for (int i = 0; i < 5; i++) {
                String orderKey = "orderKey_" + i;
                org.springframework.messaging.Message<String> message = MessageBuilder.withPayload("hello " + orderKey)
                        .setHeader("key", orderKey).build();

                // destination 格式为 topic:tag 指定好一级分类和二级分类
                // message Message 实体 你需要发送的消息体
                // timeout 消息发送超时时间
                // delayLevel 延时等级, 共有18个延时等级可选择
                // 注意: 同步异步发送消息都是有消息结果返回的
                TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("TOPIC_TEST_TRANS", message, orderKey);
                System.out.println("transSend orderKey result:" + result);
                // 发送状态
                String name = result.getSendStatus().name();
                System.out.println("transSend orderKey sendStatus:" + name);

                String localName = result.getLocalTransactionState().name();
                System.out.println("transSend orderKey localName:" + localName);
                // 需要创建消息监听器，监听本地事务执行的状态和检查本地事务状态
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 延时消息
     *		投递等级（delay level）	延迟时间	    投递等级（delay level）	延迟时间
     * 		1	                    1s		    10				        6min
     * 		2				        5s		    11				        7min
     * 		3				        10s		    12				        8min
     * 		4				        30s		    13				        9min
     * 		5				        1min		14				        10min
     * 		6				        2min		15				        20min
     * 		7				        3min		16				        30min
     * 		8				        4min		17				        1h
     * 		9				        5min		18				        2h
     *
     */
    @RequestMapping("/delaySend")
    public void delaySend() {
        try {
            for (int i = 0; i < 5; i++) {
                String orderKey = "orderKey_" + i;
                Message message = new Message();
                message.setTopic("TOPIC_TEST");
                message.setBody((orderKey + "hello !!!").getBytes());
                message.setDelayTimeLevel(5);
                SendResult sendResult = defaultMQProducer.send(message);
                System.out.println("delaySend result:" + sendResult);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 将消息打包成 Collection<Message> msgs 传入方法中即可
     * 需要注意的是批量消息的大小不能超过 1MiB（否则需要自行分割），其次同一批 batch 中 topic 必须相同
     * 消费者 只会是 同一个消费者组中 的某一个消费者
     */
    @RequestMapping("/batchSend")
    public void batchSend() {
        try {
            List<Message> messageList = new ArrayList<>();
            for (int i = 0; i < 100; i++) {
                String orderKey = "orderKey_batch " + i;
                Message message = new Message();
                message.setTopic("TOPIC_TEST");
                message.setBody((orderKey + "hello !!!").getBytes());
                messageList.add(message);
            }
            SendResult sendResult = defaultMQProducer.send(messageList);
            System.out.println("batchSend result:" + sendResult);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 广播模式， 需要在消费者监听器 中 修改 messageModel 为广播模式
     */
    @RequestMapping("/allSend")
    public void allSend() {
        try {
            for (int i = 0; i < 10; i++) {
                String orderKey = "orderKey_batch " + i;
                Message message = new Message();
                message.setTopic("TOPIC_TEST");
                message.setBody((orderKey + "hello !!!").getBytes());
                SendResult sendResult = defaultMQProducer.send(message);
                System.out.println("allSend result:" + sendResult);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 模拟死信队列 重消费
     */
    @RequestMapping("/dlqSend")
    public void dlqSend() {
        try {
//            for (int i = 0; i < 5; i++) {
//                String orderKey = "orderKey_dlq_" + i;
                String orderKey = "orderKey_dlq_";
                Message message = new Message();
                message.setTopic("TOPIC_TEST");
                message.setBody((orderKey + "hello !!!").getBytes());
                SendResult sendResult = defaultMQProducer.send(message);
                System.out.println("dlqSend result:" + sendResult);
//            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}
